{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "# `Reaqtor.IoT`\n",
    "\n",
    "Notebook equivalent of the Playground console application."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## Reference the library\n",
    "\n",
    "We'll just import the entire console application to get the transitive closure of referenced assemblies."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "#r \"bin/Debug/net6.0/Reaqtor.IoT.dll\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## (Optional) Attach a debugger\n",
    "\n",
    "If you'd like to step through the source code of the library while running samples, run the following cell, and follow instructions to start a debugger (e.g. Visual Studio). Navigate to the source code of the library to set breakpoints."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "System.Diagnostics.Debugger.Launch();"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## Import some namespaces"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "using System;\n",
    "using System.Linq;\n",
    "using System.Linq.CompilerServices.TypeSystem;\n",
    "using System.Linq.Expressions;\n",
    "using System.Threading;\n",
    "using System.Threading.Tasks;\n",
    "\n",
    "using Nuqleon.DataModel;\n",
    "\n",
    "using Reaqtive;\n",
    "using Reaqtive.Scheduler;\n",
    "\n",
    "using Reaqtor;\n",
    "using Reaqtor.IoT;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## Configure environment\n",
    "\n",
    "**Query engines** host reactive artifacts, e.g. subscriptions, which can be stateful.\n",
    "\n",
    "Query engines are a failover unit. State for all artifacts is persisted via checkpointing.\n",
    "\n",
    "Query engines depend on services from the environment:\n",
    "\n",
    "* A **scheduler** to process events on:\n",
    "  * There's one physical scheduler per host. Think of it as a thread pool.\n",
    "  * Each engine has a logical scheduler. Think of it as a collection of tasks. The engine suspends/resumes all work for checkpoint/recovery.\n",
    "* A **key/value store** for state persistence, including:\n",
    "  * A transaction log of create/delete operations for reactive artifacts.\n",
    "  * Periodic checkpoint state, which includes:\n",
    "    * State of reactive artifacts (e.g. sum and count for an Average operator).\n",
    "    * Watermarks for ingress streams, enabling replay of events upon failover.\n",
    "\n",
    "This sample also parameterizes query engines on an ingress/egress manager to receive/send events across the engine/environment boundary."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "To run query engines in the notebook, we write a simple `WithEngine` helper. This part takes care of setting up the engine and creating the environment. The general lifecycle of an engine is as follows.\n",
    "\n",
    "* Instantiate the object, passing the environment services.\n",
    "* Recover the engine's state from the key/value store.\n",
    "* Use the engine (through the `action` callback in the helper).\n",
    "* Checkpoint the engine's state. This is typically done periodically, e.g. once per minute. The interval is a tradeoff between:\n",
    "  * I/O frequency versus I/O size, e.g. due to state growth as events get processed.\n",
    "  * Replay capacity for ingress events and duration of replay, e.g. having to replay up to 1 minute worth of events from a source.\n",
    "* Unloading the engine. This is optional but useful for graceful shutdown. In the Reactor service this is used when a primary moves to another node in the cluster. It allows reactive artifacts to unload resources (e.g. connections)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "var store = new InMemoryKeyValueStore();\n",
    "var iemgr = new IngressEgressManager();\n",
    "\n",
    "async Task WithEngine(Func<MiniQueryEngine, Task> action)\n",
    "{\n",
    "    using var ps = PhysicalScheduler.Create();\n",
    "    using var scheduler = new LogicalScheduler(ps);\n",
    "    using var engine = new MiniQueryEngine(new Uri(\"iot://reactor/1\"), scheduler, store, iemgr);\n",
    "\n",
    "    using (var reader = store.GetReader())\n",
    "        await engine.RecoverAsync(reader);\n",
    "\n",
    "    await action(engine);\n",
    "\n",
    "    using (var writer = store.GetWriter())\n",
    "        await engine.CheckpointAsync(writer);\n",
    "\n",
    "    await engine.UnloadAsync();\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## Define artifacts\n",
    "\n",
    "Illustrates populating the registry of defined artifacts in the engine. This is a one-time step for the environment creating a new engine.\n",
    "\n",
    "* Artifact types that are defined include:\n",
    "  * Observables, e.g. sources of events, or query operators.\n",
    "  * Observers, e.g. sinks for events, or event handlers.\n",
    "  * Stream factories, not shown here. Useful for creation of \"subjects\" local to the engine.\n",
    "  * Subscription factories, not shown here. Useful for \"templates\" to create subscriptions with parameters.\n",
    "* All Reactor artifacts use URIs for naming purposes.\n",
    "\n",
    "The key take-away is that Reactor engines are empty by default and have no built-in artifacts whatsoever. The environment controls the registry, which includes standard query operators, specialized query operators, etc.\n",
    "\n",
    "> **Note:** There's an alternative approach to having artifacts defined in and persisted by individual engine instances. The engine can also be parameterized on a queryable external catalog. This is useful for homogeneous environments."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "await WithEngine(async engine =>\n",
    "{\n",
    "    var ctx = new ReactorContext(engine);\n",
    "\n",
    "    await ctx.DefineObserverAsync(new Uri(\"iot://reactor/observers/cout\"), ctx.Provider.CreateQbserver<T>(Expression.New(typeof(ConsoleObserver<T>))), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<TimeSpan, DateTimeOffset>(new Uri(\"iot://reactor/observables/timer\"), t => new TimerObservable(t).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "});"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## Define query operators\n",
    "\n",
    "Illustration of defining query operators, similar to defining other artifacts higher up. A few remarks:\n",
    "\n",
    "- No operators are built-in. Below, we define essential operators like `Where`, `Select`, and `Take`. The URI for these is not even prescribed; the environment picks those.\n",
    "- Implementations of the operators are provided in `Reaqtive`, similar to `System.Reactive` for classic Rx. The difference is mainly due to support for state persistence, which classic Rx lacks.\n",
    "- Custom operators are as first-class as \"standard query operators\". That is, the query engine does not have an opinion about the operator surface provided.\n",
    "\n",
    "Some ugly technicalities show up below, but those are entirely irrelevant to the user experience. The code below is part of the one-time setup provided by the environment. In particular:\n",
    "\n",
    "- Define operations are done through `IReactiveProxy`, but could also be done straight on the engine (though it brings some additional complexity when doing so).\n",
    "- There's some conversion friction to build expressions that fit through a \"queryable\" expression-tree based API but eventually bind to types in Reaqtive. That's all the As* stuff below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "await WithEngine(async engine =>\n",
    "{\n",
    "    var ctx = new ReactorContext(engine);\n",
    "\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, bool>, T>(new Uri(\"iot://reactor/observables/filter\"), (source, predicate) => source.AsSubscribable().Where(predicate).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, int, bool>, T>(new Uri(\"iot://reactor/observables/filter/indexed\"), (source, predicate) => source.AsSubscribable().Where(predicate).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, R>, R>(new Uri(\"iot://reactor/observables/map\"), (source, selector) => source.AsSubscribable().Select(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, int, R>, R>(new Uri(\"iot://reactor/observables/map/indexed\"), (source, selector) => source.AsSubscribable().Select(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, int, T>(new Uri(\"iot://reactor/observables/take\"), (source, count) => source.AsSubscribable().Take(count).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "});"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## Ingress and egress\n",
    "\n",
    "Illustration of defining ingress/egress proxies as observable/observer artifacts.\n",
    "\n",
    "Also see the implementation of `IngressObservable<T>` and `EgressObserver<T>`, which use the ingress/egress manager to connect to the outside world. The essence is this:\n",
    "\n",
    "- To the query running inside the engine, these look like ordinary Rx artifacts implemented using interfaces base classes provided by Reactor:\n",
    "  - `ISubscribable<T>` rather than `IObservable<T>`, to support the richer lifecycle of artifacts in Reactor compared to Rx.\n",
    "  - `Load`/`Save` state operations for checkpointing.\n",
    "- The external world communicates with the engine using a variant of the observable/observer interfaces, namely `IReliable*<T>`:\n",
    "  - Events received and produced have sequence numbers.\n",
    "  - Subscription handles to receive events from the outside world have additional operations:\n",
    "    - `Start(long)` to replay events from the given sequence number.\n",
    "    - `AcknowledgeRange(long)` to allow the external service to (optionally) prune events that are no longer needed by the engine.\n",
    "- Proxies in the engine use the sequence number to provide reliability:\n",
    "  - `Save` persists the latest received sequence number. `Load` gets it back.\n",
    "  - Upon restart of an ingress proxy, the restored sequence number is used to ask for replay of events.\n",
    "  - Upon a successful checkpoint, the latest received sequence number is acknowledged to the source (allowing pruning).\n",
    "\n",
    "The Reactor service implements such ingress/egress mechanisms using services like EventHub."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "await WithEngine(async engine =>\n",
    "{\n",
    "    var ctx = new ReactorContext(engine);\n",
    "\n",
    "    await ctx.DefineObserverAsync<string, T>(new Uri(\"iot://reactor/observers/egress\"), stream => new EgressObserver<T>(stream).AsAsyncQbserver(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<string, T>(new Uri(\"iot://reactor/observables/ingress\"), stream => new IngressObservable<T>(stream).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "});"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## Define more query operators\n",
    "\n",
    "Illustrates the definition of higher-order operators such as SelectMany and GroupBy which operate on sequences of sequences (IObservable<IObservable<T>>) which is one of the most powerful aspects of Rx."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "await WithEngine(async engine =>\n",
    "{\n",
    "    var ctx = new ReactorContext(engine);\n",
    "\n",
    "    // Average\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<int>, double>(new Uri(\"iot://reactor/observables/average/int32\"), source => source.AsSubscribable().Average().AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<long>, double>(new Uri(\"iot://reactor/observables/average/int64\"), source => source.AsSubscribable().Average().AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<double>, double>(new Uri(\"iot://reactor/observables/average/double\"), source => source.AsSubscribable().Average().AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, int>, double>(new Uri(\"iot://reactor/observables/average/selector/int32\"), (source, selector) => source.AsSubscribable().Average(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, long>, double>(new Uri(\"iot://reactor/observables/average/selector/int64\"), (source, selector) => source.AsSubscribable().Average(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, double>, double>(new Uri(\"iot://reactor/observables/average/selector/double\"), (source, selector) => source.AsSubscribable().Average(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "\n",
    "    // DistinctUntilChanged\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, T>(new Uri(\"iot://reactor/observables/distinct\"), source => source.AsSubscribable().DistinctUntilChanged().AsAsyncQbservable(), null, CancellationToken.None);\n",
    "\n",
    "    // SelectMany\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, ISubscribable<R>>, R>(new Uri(\"iot://reactor/observables/bind\"), (source, selector) => source.AsSubscribable().SelectMany(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "\n",
    "    // Window\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, TimeSpan, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/hopping/time\"), (source, duration) => source.AsSubscribable().Window(duration).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, int, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/hopping/count\"), (source, count) => source.AsSubscribable().Window(count).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, TimeSpan, TimeSpan, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/sliding/time\"), (source, duration, shift) => source.AsSubscribable().Window(duration, shift).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, int, int, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/sliding/count\"), (source, count, skip) => source.AsSubscribable().Window(count, skip).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, TimeSpan, int, ISubscribable<T>>(new Uri(\"iot://reactor/observables/window/ferry\"), (source, duration, count) => source.AsSubscribable().Window(duration, count).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "\n",
    "    // GroupBy\n",
    "    await ctx.DefineObservableAsync<IAsyncReactiveQbservable<T>, Func<T, R>, IGroupedSubscribable<R, T>>(new Uri(\"iot://reactor/observables/group\"), (source, selector) => source.AsSubscribable().GroupBy(selector).AsAsyncQbservable(), null, CancellationToken.None);\n",
    "});"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## Entity types\n",
    "\n",
    "Reactor Core is built to be flexible with regards to data models, but the default data model that's well-supported originates from a graph database effort in Bing that predates Reactor. The `[Mapping]` attributes below are the means to annotate properties. These property names are used to normalize entity types in the serialized expression representation, so the query is not dependent on a concrete type in an assembly, thus allowing the structure of data types (here to represent events) to be serialized across machine boundaries without deployment of binaries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "public class SensorReading\n",
    "{\n",
    "    [Mapping(\"iot://sensor/reading/room\")]\n",
    "    public string Room { get; set; }\n",
    "\n",
    "    [Mapping(\"iot://sensor/reading/temperature\")]\n",
    "    public double Temperature { get; set; }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "## A temperature simulator\n",
    "\n",
    "Add other streams to connect to the environment, simulating a temperature sensor reading and a feedback channel to control an A/C unit."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "var readings = iemgr.CreateSubject<SensorReading>(\"bart://sensors/home/livingroom/temperature/readings\");\n",
    "var settings = iemgr.CreateSubject<double?>(\"bart://sensors/home/livingroom/temperature/settings\");"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "Next, we define a few constants for the simulation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "var rand = new Random();\n",
    "\n",
    "//\n",
    "// Speed and granularity of simulation.\n",
    "//\n",
    "var timeStep = TimeSpan.FromMinutes(15);\n",
    "var simulationDelay = TimeSpan.FromMilliseconds(250);\n",
    "\n",
    "//\n",
    "// Absolute value of temperature gain/loss per unit time of the house adjusting to the outside temperature.\n",
    "//\n",
    "var insulationTemperatureIncrement = 0.1;\n",
    "\n",
    "//\n",
    "// Absolute value of temperature gain/loss per unit time due to the A/C unit cooling down or heating up.\n",
    "//\n",
    "var acTemperatureIncrement = 0.2;\n",
    "\n",
    "//\n",
    "// Temperature sensitivity of the thermostat to trigger turning off the A/C unit, i.e. within this range from target.\n",
    "//\n",
    "var thermostatSensitivity = 0.5;\n",
    "\n",
    "//\n",
    "// Configuration of simulation: minimum and maximum temperature outside, and coldest time of day.\n",
    "//\n",
    "var outsideMin = 55;\n",
    "var outsideMax = 85;\n",
    "var coldestTime = new TimeSpan(5, 0, 0); // 5AM\n",
    "\n",
    "//\n",
    "// Scale for the temperature range, to multiply [0..1] by to obtain a temperature value that can be added to the minimum.\n",
    "//\n",
    "var scale = outsideMax - outsideMin;\n",
    "\n",
    "//\n",
    "// Offset to the midpoint of the temperature range. Outside temperature will vary as a sine wave around this value.\n",
    "//\n",
    "var offset = outsideMin + scale / 2;\n",
    "\n",
    "#pragma warning disable CA5394 // Do not use insecure randomness. (Okay for simulation purposes.)\n",
    "\n",
    "//\n",
    "// Random initial value inside, within the range of temperatures.\n",
    "//\n",
    "var inside = outsideMin + rand.NextDouble() * scale;\n",
    "\n",
    "#pragma warning restore CA5394\n",
    "\n",
    "//\n",
    "// null if A/C unit is off; otherwise, target temperature.\n",
    "//\n",
    "var target = default(double?);\n",
    "\n",
    "//\n",
    "// Clock driven by the simulation.\n",
    "//\n",
    "var time = DateTime.Today;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "javascript"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "kernel.root.findKernelByName(\"javascript\").registerCommandHandler({commandType: 'SmartHomeCommand', handle: c => {\n",
    "    console.log(c.commandEnvelope);\n",
    "    \n",
    "    let visualisation = window.frames.visualisation; \n",
    "    \n",
    "    if (visualisation)\n",
    "    {\n",
    "        visualisation.postMessage(c.commandEnvelope.command, '*');\n",
    "    }\n",
    "}});"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "using Microsoft.DotNet.Interactive;\n",
    "using Microsoft.DotNet.Interactive.Commands;\n",
    "\n",
    "public class SmartHomeCommand : KernelCommand\n",
    "{\n",
    "    public SmartHomeCommand(): base(\"javascript\") {}\n",
    "    public DateTime? Timestamp {get; set; }\n",
    "    public Thermostat Thermostat {get; set; }\n",
    "    public Temperature Temperature {get; set; }\n",
    "    public ReaqtorStatus Reaqtor {get; set; }\n",
    "}\n",
    "\n",
    "public class Temperature \n",
    "{\n",
    "    public double Inside {get; set; }\n",
    "    public double Outside {get; set; }\n",
    "}\n",
    "\n",
    "public class Thermostat\n",
    "{\n",
    "    public string State {get; set; }\n",
    "    public string Mode {get; set; }\n",
    "}\n",
    "\n",
    "public class ReaqtorStatus\n",
    "{\n",
    "    public ReaqtorState State {get; set; } = ReaqtorState.Off;\n",
    "}\n",
    "\n",
    "public enum ReaqtorState\n",
    "{\n",
    "    Off = 0,\n",
    "    Starting = 1,\n",
    "    Crashing = 2,\n",
    "    FailingOver = 3,\n",
    "    Recovered = 4,\n",
    "    ShuttingDownGracefully = 5,\n",
    "    CreatingSubscription = 6,\n",
    "    DisposingSubscription = 7,\n",
    "    Running = 8,\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "var jsKernel = Kernel.Root.FindKernelByName(\"javascript\");\n",
    "jsKernel.RegisterCommandType<SmartHomeCommand>();"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "Now we can write a simulator routine that will generate `reading`, and set up a subscription to the `settings` stream to show the results emitted by the Reaqtor query we'll construct later."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "IDisposable SubscribeToSettingsStream()\n",
    "{\n",
    "    //\n",
    "    // Print commands arriving at thermostat.\n",
    "    //\n",
    "    return settings.Subscribe(Observer.Create<(long sequenceId, double? item)>(s =>\n",
    "    {\n",
    "        target = s.item;\n",
    "       \n",
    "        Console.WriteLine($\"STSS: {time} thermostat> {(target == null ? \"OFF\" : \"ON \" + (target > inside ? \"heating\" : \"cooling\") + \" to \" + target)}\");\n",
    "\n",
    "         var task = jsKernel.SendAsync(\n",
    "             new SmartHomeCommand \n",
    "             {\n",
    "               Timestamp = time,\n",
    "               Thermostat = new Thermostat \n",
    "               {\n",
    "                 State = (target == null ? \"OFF\" : \"ON\"),\n",
    "                 Mode = (target > inside ? \"heating\" : \"cooling\")\n",
    "               }\n",
    "           });\n",
    "    }));\n",
    "}\n",
    "\n",
    "Task RunReadingsGenerator(CancellationToken token)\n",
    "{\n",
    "    //\n",
    "    // Run simulation which adjusts both inside and outside temperature.\n",
    "    //\n",
    "    return Task.Run(async () =>\n",
    "    {\n",
    "        while (!token.IsCancellationRequested)\n",
    "        {\n",
    "            var now = (time.TimeOfDay - coldestTime - TimeSpan.FromHours(6)).TotalSeconds;\n",
    "            var secondsPerDay = TimeSpan.FromHours(24).TotalSeconds;\n",
    "\n",
    "            var outside = scale * Math.Sin(2 * Math.PI * now / secondsPerDay) / 2 + offset;\n",
    "\n",
    "            var environmentEffect = outside < inside ? -insulationTemperatureIncrement : insulationTemperatureIncrement;\n",
    "            var acUnitEffect = target != null ? (target < inside ? -acTemperatureIncrement : acTemperatureIncrement) : 0.0;\n",
    "\n",
    "            inside += environmentEffect + acUnitEffect;\n",
    "\n",
    "            if (target != null && Math.Abs(target.Value - inside) < thermostatSensitivity)\n",
    "            {\n",
    "                target = null;\n",
    "            }\n",
    "\n",
    "            await jsKernel.SendAsync(\n",
    "                new SmartHomeCommand \n",
    "                {\n",
    "                  Timestamp = time,\n",
    "                  Temperature = new Temperature \n",
    "                  {\n",
    "                    Inside = inside,\n",
    "                    Outside = outside\n",
    "                  }\n",
    "              });\n",
    "\n",
    "            // Console.WriteLine($\"RRG: {time} temperature> inside = {inside} outside = {outside} target = {target}\");\n",
    "            readings.OnNext((Environment.TickCount, new SensorReading { Room = \"Hallway\", Temperature = inside }));\n",
    "\n",
    "            await Task.Delay(simulationDelay);\n",
    "            time += timeStep;\n",
    "        }\n",
    "    });\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "html"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "<!DOCTYPE html>\n",
    "<html>\n",
    "<head>\n",
    "    <title></title>\n",
    "    <meta charset=\"utf-8\" />\n",
    "</head>\n",
    "<body>\n",
    "    <iframe name=\"visualisation\" src=\"https://reaqtor-house.netlify.app/?auto=false\" width=\"100%\" height=\"500\"></iframe>\n",
    "</body>\n",
    "</html>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "polyglot_notebook": {
     "kernelName": "csharp"
    }
   },
   "source": [
    "In the next cell, we'll write a higher-order query using `Window` and `SelectMany`, and run the simulator and logger while the query is running."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "polyglot-notebook"
    }
   },
   "outputs": [],
   "source": [
    "var stopEventProducer = new CancellationTokenSource();\n",
    "\n",
    "Console.WriteLine(\"Starting simulator for temperature sensor readings...\");\n",
    "\n",
    "var logger = SubscribeToSettingsStream();\n",
    "var producer = RunReadingsGenerator(stopEventProducer.Token);\n",
    "\n",
    "var subUri = new Uri(\"iot://reactor/subscription/BD/livingroom/comfy\");\n",
    "\n",
    "Console.WriteLine(\"Setting up query engine...\");\n",
    "\n",
    "await jsKernel.SendAsync(new SmartHomeCommand { Reaqtor = new ReaqtorStatus { State = ReaqtorState.Starting }});\n",
    "\n",
    "await WithEngine(async engine =>\n",
    "{\n",
    "    var ctx = new ReactorContext(engine);\n",
    "\n",
    "    var input = ctx.GetObservable<string, SensorReading>(new Uri(\"iot://reactor/observables/ingress\"));\n",
    "    var output = ctx.GetObserver<string, double?>(new Uri(\"iot://reactor/observers/egress\"));\n",
    "\n",
    "    var readings = input(\"bart://sensors/home/livingroom/temperature/readings\");\n",
    "    var settings = output(\"bart://sensors/home/livingroom/temperature/settings\");\n",
    "\n",
    "    Console.WriteLine(\"Creating subscription...\");\n",
    "    await jsKernel.SendAsync(new SmartHomeCommand { Reaqtor = new ReaqtorStatus { State = ReaqtorState.CreatingSubscription }});\n",
    "\n",
    "    await readings.Window(4).SelectMany(w => w.Average(r => r.Temperature)).Select(t => t < 70 || t > 80 ? 75 : default(double?)).DistinctUntilChanged().SubscribeAsync(settings, subUri, null, CancellationToken.None);\n",
    "\n",
    "    await jsKernel.SendAsync(new SmartHomeCommand { Reaqtor = new ReaqtorStatus { State = ReaqtorState.Running }});\n",
    "\n",
    "    // Run for a bit.\n",
    "    await Task.Delay(TimeSpan.FromSeconds(30));\n",
    "\n",
    "    await jsKernel.SendAsync(new SmartHomeCommand { Reaqtor = new ReaqtorStatus { State = ReaqtorState.FailingOver }});\n",
    "\n",
    "    Console.WriteLine(\"Engine failing over... Note we'll continue to see the producer's `temperature>` traces, but no `thermostat>` outputs.\");\n",
    "});\n",
    "\n",
    "await Task.Delay(TimeSpan.FromSeconds(2));\n",
    "\n",
    "await WithEngine(async engine =>\n",
    "{\n",
    "    await jsKernel.SendAsync(new SmartHomeCommand { Reaqtor = new ReaqtorStatus { State = ReaqtorState.Recovered }});\n",
    "    Console.WriteLine(\"Engine recovered!\");\n",
    "\n",
    "    var ctx = new ReactorContext(engine);\n",
    "\n",
    "    await jsKernel.SendAsync(new SmartHomeCommand { Reaqtor = new ReaqtorStatus { State = ReaqtorState.Running }});\n",
    "    \n",
    "    // Run for a bit more.\n",
    "    await Task.Delay(TimeSpan.FromSeconds(30));\n",
    "\n",
    "    await jsKernel.SendAsync(new SmartHomeCommand { Reaqtor = new ReaqtorStatus { State = ReaqtorState.DisposingSubscription }});\n",
    "    Console.WriteLine(\"Disposing subscription...\");\n",
    "\n",
    "    await ctx.GetSubscription(subUri).DisposeAsync();\n",
    "});\n",
    "\n",
    "await jsKernel.SendAsync(new SmartHomeCommand { Reaqtor = new ReaqtorStatus { State = ReaqtorState.ShuttingDownGracefully }});\n",
    "Console.WriteLine(\"Stopping simulator...\");\n",
    "\n",
    "stopEventProducer.Cancel();\n",
    "producer.Wait();\n",
    "\n",
    "logger.Dispose();\n",
    "\n",
    "await jsKernel.SendAsync(new SmartHomeCommand { Reaqtor = new ReaqtorStatus { State = ReaqtorState.Off }});\n",
    "\n",
    "Console.WriteLine(\"Done!\");"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".NET (C#)",
   "language": "C#",
   "name": ".net-csharp"
  },
  "polyglot_notebook": {
   "kernelInfo": {
    "defaultKernelName": "csharp",
    "items": [
     {
      "aliases": [
       "c#",
       "C#"
      ],
      "languageName": "C#",
      "name": "csharp"
     },
     {
      "aliases": [],
      "name": ".NET"
     },
     {
      "aliases": [
       "f#",
       "F#"
      ],
      "languageName": "F#",
      "name": "fsharp"
     },
     {
      "aliases": [],
      "languageName": "HTML",
      "name": "html"
     },
     {
      "aliases": [],
      "languageName": "KQL",
      "name": "kql"
     },
     {
      "aliases": [],
      "languageName": "Mermaid",
      "name": "mermaid"
     },
     {
      "aliases": [
       "powershell"
      ],
      "languageName": "PowerShell",
      "name": "pwsh"
     },
     {
      "aliases": [],
      "languageName": "SQL",
      "name": "sql"
     },
     {
      "aliases": [],
      "name": "value"
     },
     {
      "aliases": [
       "frontend"
      ],
      "name": "vscode"
     },
     {
      "aliases": [
       "js"
      ],
      "languageName": "JavaScript",
      "name": "javascript"
     }
    ]
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
